/*
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 * 
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Library General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor Boston, MA 02110-1301,  USA
 */

/**
 * @version conn.c 0.0.2
 * @package Comoro
 * @author Dr.NP <np@bsgroup.org>
 *
 * Network connection service handler
 * Based on libevent
 */

#include "bspd.h"

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <errno.h>
#include <signal.h>
#include <fcntl.h>
#include <time.h>
#include <sys/resource.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <event.h>

#ifdef HAVE_MALLOC_H
#ifndef __OpenBSD__
#include <malloc.h>
#endif
#endif

struct conn_client **client_list;
struct conn_server *server_list[SERVER_POOL_SIZE];
struct conn_connector *connector_list[CONNECTOR_POOL_SIZE];
extern struct event_base *ev_base;
static int client_list_size;
static int total_clients;
static int max_clients;

/* Initialization of socket object */
int conn_init()
{
	// Initialize object pool
	fprintf(stderr, "Socket connector initialized ... ");
	conn_client_list_init();
	memset(server_list, 0, sizeof(struct conn_server *) * SERVER_POOL_SIZE);
	memset(connector_list, 0, sizeof(struct conn_connector *) * CONNECTOR_POOL_SIZE);

	queue_clock_cb_func_add(conn_client_hb_dt, NULL);

	return RTN_SUCCESS;
}

/* Close all client from connector */
int conn_close()
{
	int i;
	
	for (i = 0; i < client_list_size; i ++)
	{
		conn_client_free(client_list[i]);
	}

	for (i = 0; i < CONNECTOR_POOL_SIZE; i ++)
	{
		conn_tcp_connector_free(connector_list[i]);
	}

	for (i = 0; i < SERVER_POOL_SIZE; i ++)
	{
		conn_tcp_server_free(server_list[i]);
	}
	
	return RTN_SUCCESS;
}

/* Set non-blocking */
int conn_setnonblock(int fd)
{
	int flags;
	
	flags = fcntl(fd, F_GETFL);
	if (flags < 0)
	{
		return flags;
	}
	
	flags |= O_NONBLOCK;
	if (fcntl(fd, F_SETFL, flags) < 0)
	{
		return -1;
	}
	
	return 0;
}

/* Generate a TCP server by given struct */
int conn_tcp_server_new(struct conn_server *c)
{
	int id = conn_server_list_find_free();
	
	if (id < 0)
	{
		fprintf(stderr, "Server pool full, cannot create more tcp server\n");
		log_add("Server pool full");

		return RTN_GENERAL;
	}
	
	if (c->srv_addr < 0)
	{
		c->srv_addr = INADDR_ANY;
	}

	if (c->srv_port > 65535)
	{
		return RTN_GENERAL;
	}

	// If port number below 1024, root privilege needed
	if (c->srv_port <= 1024)
	{
		// CHECK ROOT PRIVILEGE
		if (0 != getuid() && 0 != geteuid())
		{
			fprintf(stderr, "Port number below 1024, root privilege needed\n");
			return RTN_GENERAL;
		}
	}

	// Create new socket
	c->listen_fd = socket(AF_INET, SOCK_STREAM, 0);
	if (c->listen_fd < 0)
	{
		fprintf(stderr, "Socket creation failed\n");
		exit(EXIT_SOCKET_CREATE_ERROR);
	}

	c->listen_addr.sin_family = AF_INET;
	c->listen_addr.sin_addr.s_addr = c->srv_addr;
	c->listen_addr.sin_port = htons(c->srv_port);

	// Bind socket
	if (bind(c->listen_fd, (struct sockaddr *) &c->listen_addr, sizeof(c->listen_addr)) < 0)
	{
		fprintf(stderr, "Socket bind failed on port %lu:%d %s\n", c->srv_addr, c->srv_port, strerror(errno));
		exit(EXIT_SOCKET_BIND_ERROR);
	}

	// Listen socket
	if (listen(c->listen_fd, c->listen_queue_length) < 0)
	{
		fprintf(stderr, "Socket listen failed: %s\n", strerror(errno));
		exit(EXIT_SOCKET_LISTEN_ERROR);
	}

	// Set socket options
	struct timeval timeout = {.tv_sec = c->tcp_timeout, .tv_usec = 0};
	
	setsockopt(c->listen_fd, SOL_SOCKET, SO_REUSEADDR, &c->tcp_reuse, sizeof(c->tcp_reuse));
	setsockopt(c->listen_fd, SOL_SOCKET, SO_SNDTIMEO, (char *) &timeout, sizeof(struct timeval));
	setsockopt(c->listen_fd, SOL_SOCKET, SO_SNDBUF, &c->tcp_send_buffer_size, sizeof(c->tcp_send_buffer_size));
	setsockopt(c->listen_fd, SOL_SOCKET, SO_RCVBUF, &c->tcp_recv_buffer_size, sizeof(c->tcp_recv_buffer_size));
	setsockopt(c->listen_fd, IPPROTO_TCP, TCP_NODELAY, (char *) &c->tcp_nodelay, sizeof(c->tcp_nodelay));

	// Set socket non-blocking
	if (c->nonblock && conn_setnonblock(c->listen_fd) < 0)
	{
		fprintf(stderr, "Socket set non-blocking failed\n");
		exit(EXIT_SOCKET_NONBLOCK_ERROR);
	}

	c->start_time = time((time_t *) NULL);

	// Add to server pool
	c->id = id;
	server_list[id] = c;
	
	fprintf(stderr, "Socket server created on port %d\n", c->srv_port);
	log_add("Socket server created on port %d", c->srv_port);

	// Add data handler
	event_set(&c->ev_accept, c->listen_fd, EV_READ|EV_PERSIST, conn_tcp_server_accept, (void *) c);
	event_base_set(ev_base, &c->ev_accept);
	event_add(&c->ev_accept, NULL);
	
	return c->listen_fd;
}

/* Accept new connection */
void conn_tcp_server_accept(int fd, short ev, void *arg)
{
	int client_fd;
	struct sockaddr_in client_addr;
	socklen_t client_len = sizeof(client_addr);
	struct conn_client *client;
	struct conn_server *server = (struct conn_server *) arg;

	// Accept new connection
	client_fd = accept(fd, (struct sockaddr *) &client_addr, &client_len);
	if (-1 == client_fd)
	{
		fprintf(stderr, "Accept error\n");
		log_add("Accept error");
		return;
	}
	
	// Add connection to event queue
	client = conn_client_new(client_fd);
	if (!client)
	{
		fprintf(stderr, "Server limited: I cannot serve more clients\n");
		log_add("Server Limit");
		return;
	}

	// Try to set non-blocking
	if (conn_setnonblock(client_fd) < 0)
	{
		conn_client_free(client);
		fprintf(stderr, "Client socket set non-blocking error\n");
		log_add("Client %d set non-blocking failed", client->id);
	}
	
	// Client initialize
	client->server_id = server->id;
	client->client_addr = inet_addr(inet_ntoa(client_addr.sin_addr));
	client->client_port = client_addr.sin_port;
	client->conn_time = time((time_t *) NULL);
	client->last_hb_time = client->conn_time;

	if (server->on_conn_callback)
	{
		server->on_conn_callback(client->id);
	}
	
	thread_worker_dispatch(WORKER_TYPE_CONN, client->id, WORKER_DISPATCH_CHILD_ONLY);
	
	return;
}

/* Close and free a TCP server */
int conn_tcp_server_free(struct conn_server *server)
{
	int i;
	
	if (!server)
	{
		return RTN_GENERAL;
	}

	if (server->listen_fd)
	{
		// Close all clients connected to this server
		for (i = 0; i < client_list_size; i ++)
		{
			if (client_list[i] && client_list[i]->server_id == server->id)
			{
				conn_client_free(client_list[i]);
			}
		}
		
		shutdown(server->listen_fd, 2);
		close(server->listen_fd);
	}

	memset(server, 0, sizeof(struct conn_server));

	return RTN_SUCCESS;
}

/* Event on data from client */
void conn_tcp_server_on_data(int fd, short ev, void *arg)
{
	struct conn_client *client = (struct conn_client *) arg;
	int res;
	unsigned int buffer_curr = 0;
	unsigned int buffer_avail = 0;
	char *new_buffer;
	
	if (!client || !client->client_fd)
	{
		return;
	}

	struct conn_server *server = server_list[client->server_id];
	client->buffer_data_size = 0;

	if (fd != client->client_fd)
	{
		// Sanity
		conn_client_free(client);
		return;
	}

	if (!client->read_buffer)
	{
		// Initialize buffer
		client->read_buffer = (char *) malloc(CLIENT_READ_BUFFER_SIZE);
		client->buffer_size = buffer_avail = CLIENT_READ_BUFFER_SIZE;
		memset(client->read_buffer, 0, CLIENT_READ_BUFFER_SIZE);
	}

	else
	{
		buffer_avail = client->buffer_size;
		client->buffer_data_size = 0;
	}

	while(1)
	{
		res = read(client->client_fd, client->read_buffer + buffer_curr, buffer_avail);

		if (res > 0)
		{
			if ((unsigned int) res == buffer_avail)
			{
				// Buffer full, maybe stream available, enlarge buffer
				if ((new_buffer = (char *) realloc(client->read_buffer, client->buffer_size * 2)) == NULL)
				{
					fprintf(stderr, "Enlarge read buffer error\n");
					log_add("Client %d enlarge read buffer error", client->id);
					conn_client_free(client);
					
					return;
				}
				
				buffer_avail = client->buffer_size;
				client->read_buffer = new_buffer;
				memset(client->read_buffer + client->buffer_size, 0, client->buffer_size);
				client->buffer_size *= 2;
			}
			
			else
			{
				// Data read completely
				buffer_avail -= res;
			}

			client->buffer_data_size += res;
			
			break;
		}

		if (-1 == res)
		{
			// Read error
			fprintf(stderr, "Socket read error\n");
			log_add("Client %d read error", client->id);
			conn_client_free(client);
			
			return;
		}
		
		if (0 == res && 0 == client->buffer_data_size)
		{
			// Connection closed
			conn_client_free(client);
			
			return;
		}
	}
	
	// Call receiver
	if (server->on_data_callback)
	{
		server->on_data_callback(client->id, (void *) client->read_buffer, client->buffer_data_size);
	}

	// Clear source data. if source size too big, free it~
	if (client->buffer_size >= CLIENT_READ_BUFFER_SIZE_HIGHWAT && client->read_buffer)
	{
		free(client->read_buffer);
		client->read_buffer = (char *) NULL;
	}
	
	else
	{
		if (client->read_buffer)
		{
			memset(client->read_buffer, 0, client->buffer_size);
		}
	}
	
	return;
}

/* New connection client */
struct conn_client * conn_client_new(int client_fd)
{
	int id;
	struct conn_client *tmp = (struct conn_client *) NULL;

	// Find a free slot
	id = conn_client_list_find_free ();
	if (id >= 0)
	{
		tmp = conn_client_list_get(id);

		if (!tmp)
		{
			tmp = (struct conn_client *) malloc(sizeof(struct conn_client));
		}
	}

	else
	{
		// No empty slot, enlarge pool
		id = conn_client_list_enlarge() / 2;
		if (id > 0)
		{
			tmp = (struct conn_client *) malloc(sizeof(struct conn_client));
		}
	}

	if (tmp)
	{
		memset(tmp, 0, sizeof(struct conn_client));
		conn_client_clear(tmp);
		tmp->id = id;
		tmp->client_fd = client_fd;
		client_list[id] = tmp;

		total_clients ++;
		
		if (max_clients < total_clients)
		{
			max_clients = total_clients;
		}
	}
	
	return tmp;
}

/* Empty client struct data */
int conn_client_clear(struct conn_client *client)
{
	int i;
	
	if (client)
	{
		// Clear all data
		
		client->conn_time = 0;
		client->last_hb_time = 0;
		client->client_fd = 0;
		client->server_id = -1;
		client->client_addr = 0;
		client->client_port = 0;
		client->last_rcv_id = 0;
		client->last_snd_id = 0;
		client->buffer_size = 0;
		client->runner_id = -1;
		
		if (client->read_buffer)
		{
			free(client->read_buffer);
			client->read_buffer = (char *) NULL;
		}
		
		// Initialize sequence buffer
		for (i = 0; i < SEQ_BUFFER_SIZE; i ++)
		{
			if (client->seq_buffer[i])
			{
				if (client->seq_buffer[i]->seq_str)
				{
					free(client->seq_buffer[i]->seq_str);
				}
			
				free(client->seq_buffer[i]);
				client->seq_buffer[i] = (struct sequence_item *) NULL;
			}
		}
		
		// Delete event
		event_del(&client->ev_read);

		return RTN_SUCCESS;
	}

	return RTN_MEMORY_ERROR;
}

/* Close a client and free all data */
int conn_client_free(struct conn_client *client)
{
	if (!client || !client->client_fd)
	{
		return RTN_GENERAL;
	}

	struct conn_server *server = server_list[client->server_id];
	
	shutdown(client->client_fd, 2);
	close(client->client_fd);

	conn_client_clear(client);

	if (server && server->on_close_callback)
	{
		server->on_close_callback(client->id);
	}

	total_clients --;

	return RTN_SUCCESS;
}

/* Send data to client */
int conn_client_send(struct conn_client *client, const char *msg, int msglen)
{
	int len = 0, t;

	if (!client || !client->client_fd)
	{
		return RTN_GENERAL;
	}
	
	if (msglen < 0)
	{
		msglen = strlen(msg) + 1;
	}
	
	len = t = send(client->client_fd, msg, msglen, 0);

	if (-1 == len)
	{
		log_add("Network send error to client %d on FD<%d>", client->id, client->client_fd);
		
		return len;
	}
	
	if (len < msglen)
	{
		// Send buffer full, loop until buffer empty
		while (len >= msglen)
		{
			t = send(client->client_fd, msg + len, msglen - len, 0);
			
			if (-1 == len)
			{
				// Socket error, break
				return len;
			}

			len += t;
		}
	}
	
	return len;
}

/* Set heartbeat */
void conn_client_hb(struct conn_client *client)
{
	static char signal[9] = {0xFF, 0xFF, 0xFF, 0xFF, 0x00, 0x00, 0x00, 0x00};
	
	if (client && client->conn_time > 0)
	{
		client->last_hb_time = time((time_t *) NULL);
	}

	// Send response
	conn_client_send(client, signal, 8);

	return;
}

/* Heartbeat timeout detection */
void conn_client_hb_dt(void *arg)
{
	int i;
	unsigned int now = time((time_t *) NULL);
	
	for (i = 0; i < client_list_size; i ++)
	{
		if (client_list[i] && client_list[i]->conn_time > 0)
		{
			// A live connection
			if (now - client_list[i]->last_hb_time > TCP_TIMEOUT && client_list[i]->client_addr && (client_list[i]->client_addr != inet_addr(ADMIN_IP)))
			{
				// Close it
				conn_client_free(client_list[i]);
			}
			
			else
			{
				// Flush TCP buffer
				conn_client_seq_flush(client_list[i]);
			}
		}
	}
}

/* Set current recv sequence id */
void conn_client_rcv_seq_set(struct conn_client *client, unsigned int seq_id)
{
	if (!client)
	{
		return;
	}

	client->last_rcv_id = seq_id;
	
	return;
}

/* Get current recv sequence id */
unsigned int conn_client_rcv_seq_get(struct conn_client *client)
{
	if (!client)
	{
		return 0xFFFFFFFF;
	}

	return client->last_rcv_id;
}

/* Set current recv sequence id */
void conn_client_snd_seq_set(struct conn_client *client, unsigned int seq_id)
{
	if (!client)
	{
		return;
	}

	client->last_snd_id = seq_id;
	
	return;
}

/* Get current recv sequence id */
unsigned int conn_client_snd_seq_get(struct conn_client *client)
{
	if (!client)
	{
		return 0xFFFFFFFF;
	}

	return client->last_snd_id;
}

/* Add content to sequence buffer */
unsigned int conn_client_seq_add(struct conn_client *client, char *data, int len)
{
	int i;

	if (!client)
	{
		return 0;
	}
	
	for (i = 0; i < SEQ_BUFFER_SIZE; i ++)
	{
		if (!client->seq_buffer[i])
		{
			// Found a empty buffer slot
			client->seq_buffer[i] = (struct sequence_item *) malloc(sizeof(struct sequence_item));
			client->seq_buffer[i]->seq_id = client->last_snd_id;
			client->seq_buffer[i]->seq_str = (char *) malloc(len);
			memcpy(client->seq_buffer[i]->seq_str, data, len);
			client->seq_buffer[i]->msglen = len;
			client->seq_buffer[i]->sgn = 0;
			
			//fprintf(stderr, "Sequence %u added\n", client->last_snd_id);
			
			return client->last_snd_id;
		}
	}
	
	// Buffer full, bye.... >_<!
	conn_client_free(client);
	
	return 0;
}	

/* Send sequence buffer content */
void conn_client_seq_send(struct conn_client *client, unsigned int seq_id)
{
	int i;

	if (!client)
	{
		return;
	}
	
	for (i = 0; i < SEQ_BUFFER_SIZE; i ++)
	{
		if (client->seq_buffer[i] && client->seq_buffer[i]->seq_str && client->seq_buffer[i]->seq_id == seq_id)
		{
			send(client->client_fd, client->seq_buffer[i]->seq_str, client->seq_buffer[i]->msglen, 0);
			
			return;
		}
	}
	
	return;
}

/* Delete sequence item from buffer */
void conn_client_seq_del(struct conn_client *client, unsigned int seq_id)
{
	int i;

	if (!client)
	{
		return;
	}
	
	for (i = 0; i < SEQ_BUFFER_SIZE; i ++)
	{
		if (client->seq_buffer[i] && client->seq_buffer[i]->seq_str && client->seq_buffer[i]->seq_id == seq_id)
		{
			// Clear slot
			free(client->seq_buffer[i]->seq_str);
			free(client->seq_buffer[i]);
			client->seq_buffer[i] = (struct sequence_item *) NULL;

			//fprintf(stderr, "Sequence %u deleted\n", seq_id);
		}
	}
	
	return;
}

/* Flush all sequence of a client */
int conn_client_seq_flush(struct conn_client *client)
{
	int i, ct = 0;
	
	if (!client)
	{
		return RTN_GENERAL;
	}
	
	for (i = 0; i < SEQ_BUFFER_SIZE; i ++)
	{
		if (client->seq_buffer[i] && client->seq_buffer[i]->seq_str)
		{
			if (client->seq_buffer[i]->sgn > TCP_TIMEOUT)
			{
				send(client->client_fd, client->seq_buffer[i]->seq_str, client->seq_buffer[i]->msglen, 0);
				client->seq_buffer[i]->sgn = 0;
				ct ++;
			}
			
			client->seq_buffer[i]->sgn ++;
			
			if (client->seq_buffer[i]->sgn > MAX_SEQ_RESEND)
			{
				// I shouted so many times, are you deaf???
				conn_client_free(client);
				return RTN_GENERAL;
			}
		}
	}
	
	return ct;
}

/* Flush all client's send buffer, used for clock callback */
void conn_client_seq_flush_all(void *arg)
{
	int i;
	unsigned int now = time((time_t *) NULL);
	
	for (i = 0; i < client_list_size; i ++)
	{
		if (client_list[i] && client_list[i]->conn_time > 0)
		{
			// a live connection
			if (now - client_list[i]->last_hb_time > TCP_TIMEOUT && client_list[i]->client_addr && (client_list[i]->client_addr == inet_addr(ADMIN_IP)))
			{
				// Close it
				conn_client_free(client_list[i]);
			}
			
			else
			{
				// Flush TCP buffer
				conn_client_seq_flush(client_list[i]);
			}
		}
	}
	
	return;
}

/* Make a new connection to TCP server */
int conn_tcp_connector_new(struct conn_connector *c)
{
	int id = conn_connector_list_find_free ();
	if (id < 0)
	{
		fprintf(stderr, "Connector pool full, cannot create more tcp connector\n");
		log_add("Connector pool full");

		return RTN_GENERAL;
	}
	
	// Create socket
	c->connect_fd = socket(AF_INET, SOCK_STREAM, 0);
	if (c->connect_fd < 0)
	{
		fprintf(stderr, "Socket creation failed\n");
		exit(EXIT_SOCKET_CREATE_ERROR);
	}

	c->connect_addr.sin_family = AF_INET;
	c->connect_addr.sin_addr.s_addr = c->srv_addr;
	c->connect_addr.sin_port = htons(c->srv_port);
	
	if (connect(c->connect_fd, (struct sockaddr *) &c->connect_addr, sizeof(c->connect_addr)) < 0)
	{
		fprintf(stderr, "Connect to server %lu:%d failed\n", c->srv_addr, c->srv_port);
		exit(EXIT_SOCKET_CONNECT_ERROR);
	}
	
	if (c->nonblock && conn_setnonblock(c->connect_fd) < 0)
	{
		fprintf(stderr, "Socket set non-blocking failed\n");
		exit(EXIT_SOCKET_NONBLOCK_ERROR);
	}
	
	c->conn_time = time((time_t *) NULL);

	setsockopt(c->connect_fd, SOL_SOCKET, SO_SNDBUF, &c->tcp_send_buffer_size, sizeof(c->tcp_send_buffer_size));
	setsockopt(c->connect_fd, SOL_SOCKET, SO_RCVBUF, &c->tcp_recv_buffer_size, sizeof(c->tcp_recv_buffer_size));
	setsockopt(c->connect_fd, IPPROTO_TCP, TCP_NODELAY, (char *) &c->tcp_nodelay, sizeof(c->tcp_nodelay));
	
	// Add connector to list
	c->id = id;
	connector_list[id] = c;
	
	// Data event
	event_set(&c->ev_read, c->connect_fd, EV_READ|EV_PERSIST, conn_tcp_connector_on_data, (void *) c);
	event_base_set(ev_base, &c->ev_read);
	event_add(&c->ev_read, NULL);

	// Callback
	if (c->on_conn_callback)
	{
		c->on_conn_callback();
	}

	// Add clock event - heartbeat
	queue_clock_cb_func_add(conn_tcp_connector_hb, (void *) c);

	fprintf(stderr, "Connected to server %lu:%d\n", c->srv_addr, c->srv_port);

	return c->connect_fd;
}

/* Send data from connector to connected server */
int conn_tcp_connector_send(struct conn_connector *connector, const char *msg, int msglen)
{
	int len = 0, t;

	if (!connector || !connector->connect_fd)
	{
		return RTN_GENERAL;
	}

	if (msglen < 0)
	{
		msglen = strlen(msg) + 1;
	}

	len = t = send(connector->connect_fd, msg, msglen, 0);

	if (-1 == len)
	{
		log_add("Network send error to server %d from connector <%d>", connector->srv_addr, connector->id);

		return len;
	}

	if (len < msglen)
	{
		// Send buffer full, loop until buffer empty
		while (len >= msglen)
		{
			t = send(connector->connect_fd, msg + len, msglen - len, 0);
			
			if (-1 == len)
			{
				// Socket error, break
				return len;
			}

			len += t;
		}
	}
	
	return len;
}

/* Close a connector and free memory */
int conn_tcp_connector_free(struct conn_connector *connector)
{
	if (!connector)
	{
		return RTN_GENERAL;
	}

	if (connector->connect_fd)
	{
		shutdown(connector->connect_fd, 2);
		close(connector->connect_fd);
	}

	if (connector->on_close_callback)
	{
		connector->on_close_callback();
	}

	// I FREED MEMORY!!!
	connector_list[connector->id] = (struct conn_connector *) NULL;
	if (connector->read_buffer)
	{
		free(connector->read_buffer);
	}

	event_del(&connector->ev_read);
	
	free(connector);
	
	return RTN_SUCCESS;
}

/* Event on data from connector */
void conn_tcp_connector_on_data(int fd, short ev, void *arg)
{
	struct conn_connector *connector = (struct conn_connector *) arg;
	int res;
	unsigned int buffer_avail = 0;
	unsigned int buffer_curr = 0;
	char *new_buffer;

	if (!connector || !connector->connect_fd)
	{
		return;
	}

	if (fd != connector->connect_fd)
	{
		// Sanity
		conn_tcp_connector_free(connector);
		
		return;
	}

	if (!connector->read_buffer)
	{
		// Initialize buffer
		connector->read_buffer = (char *) malloc(CONNECTOR_READ_BUFFER_SIZE);
		connector->buffer_size = buffer_avail = CONNECTOR_READ_BUFFER_SIZE;
		memset(connector->read_buffer, 0, CONNECTOR_READ_BUFFER_SIZE);
	}

	else
	{
		buffer_avail = connector->buffer_size;
		connector->buffer_data_size = 0;
	}

	while(1)
	{
		res = read(connector->connect_fd, connector->read_buffer + buffer_curr, buffer_avail);

		if (res > 0)
		{
			if ((unsigned int) res == buffer_avail)
			{
				// Buffer full, maybe stream available, enlarge buffer
				if ((new_buffer = (char *) realloc(connector->read_buffer, connector->buffer_size * 2)) == NULL)
				{
					fprintf(stderr, "Enlarge read buffer error\n");
					log_add("Connector %d enlarge read buffer error", connector->id);
					conn_tcp_connector_free(connector);
					
					return;
				}
				
				buffer_avail = connector->buffer_size;
				connector->read_buffer = new_buffer;
				memset(connector->read_buffer + connector->buffer_size, 0, connector->buffer_size);
				connector->buffer_size *= 2;
			}
			
			else
			{
				// Data read completely
				buffer_avail -= res;
			}

			connector->buffer_data_size += res;
			
			break;
		}

		if (-1 == res)
		{
			// Read error
			fprintf(stderr, "Socket read error\n");
			log_add("Connector %d read error", connector->id);
			conn_tcp_connector_free(connector);
			
			return;
		}
		
		if (0 == res && 0 == connector->buffer_data_size)
		{
			// Connection closed
			conn_tcp_connector_free(connector);
			
			return;
		}
	}

	// Call receiver
	if (connector->on_data_callback)
	{
		connector->on_data_callback((void *) connector->read_buffer, connector->buffer_data_size);
	}

	// Clear source data. if source size too big, free it~
	if (connector->buffer_size >= CONNECTOR_READ_BUFFER_SIZE_HIGHWAT && connector->read_buffer)
	{
		free(connector->read_buffer);
		connector->read_buffer = (char *) NULL;
	}
	
	else
	{
		if (connector->read_buffer)
		{
			memset(connector->read_buffer, 0, connector->buffer_size);
		}
	}
	
	return;
}

/* Connector heartbeat */
void conn_tcp_connector_hb(void *arg)
{
	struct conn_connector *connector = (struct conn_connector *) arg;
	static char signal[9] = {0xFF, 0xFF, 0xFF, 0xFF, 0x00, 0x00, 0x00, 0x00};

	conn_tcp_connector_send(connector, signal, 8);

	return;
}

/* Find a free slot from server pool */
int conn_server_list_find_free()
{
	int i;
	
	for (i = 0; i < SERVER_POOL_SIZE; i ++)
	{
		if (!server_list[i] || 0 == server_list[i]->start_time)
		{
			return i;
		}
	}

	return -1;
}

/* Initialization of client list */
int conn_client_list_init()
{
	if ((client_list = (struct conn_client **) malloc(CLIENT_POOL_BASE_SIZE * sizeof(struct conn_client *))) == NULL)
	{
		fprintf(stderr, "Malloc error, cannot init client pool\n");
		exit(EXIT_MEMORY_ERROR);
	}

	client_list_size = CLIENT_POOL_BASE_SIZE;
	memset(client_list, 0, client_list_size * sizeof(struct conn_client *));
	total_clients = 0;
	max_clients = 0;
	
	fprintf(stderr, "Client pool initialized as %d\n", CLIENT_POOL_BASE_SIZE);

	return CLIENT_POOL_BASE_SIZE;
}

/* Enlarge (double) client pool size */
int conn_client_list_enlarge()
{
	// Pool full? enlarge its size
	if (client_list_size >= MAX_CLIENT)
	{
		// Client list full, OMG~~ so you are crazy!
		log_add("Client list pool full, if you want to enlarge it, modify conn.h source please");
		fprintf(stderr, "Client list full\n");
		
		return 0;
	}
	
	struct conn_client **new_list = (struct conn_client **) realloc(client_list, sizeof(struct conn_client *) * client_list_size * 2);
	
	if (new_list)
	{
		log_add("Client_list pool enlarged from %d to %d", client_list_size, client_list_size * 2);
		
		memset(&new_list[client_list_size], 0, sizeof(struct conn_client *) * client_list_size);
		client_list = new_list;
		client_list_size *= 2;
	}

	return client_list_size;
}

/* Purge client pool */
int conn_client_list_purge()
{
	// ARE YOU CRAZY ???
	int i, t = 0;

	for (i = 0; i < client_list_size; i ++)
	{
		// Clear unused slot
		if (client_list[i] && 0 == client_list[i]->conn_time)
		{
			conn_client_clear(client_list[i]);
			free(client_list[i]);
			client_list[i] = (struct conn_client *) NULL;

			t ++;
		}
	}
	
	return t;
}

/* Find a free slot from client pool */
int conn_client_list_find_free()
{
	int i;
	
	for (i = 0; i < client_list_size; i ++)
	{
		if (!client_list[i] || 0 == client_list[i]->conn_time)
		{
			return i;
		}
	}

	return -1;
}

/* Get client object from pool by given client_id */
struct conn_client * conn_client_list_get(int id)
{
	return client_list[id];
}

/* Count clients */
int conn_client_list_count()
{
	return total_clients;
}

/* Max clients */
int conn_client_list_max()
{
	return max_clients;
}

/* Client list size */
int conn_client_list_size()
{
	return client_list_size;
}

/* Find a free slot from connector pool */
int conn_connector_list_find_free()
{
	int i;

	for (i = 0; i < CONNECTOR_POOL_SIZE; i ++)
	{
		if (!connector_list[i] || 0 == client_list[i]->conn_time)
		{
			return i;
		}
	}

	return -1;
}
